package com.mls.guardian.remoting.netty;

import com.mls.guardian.common.RemoteCommandType;
import com.mls.guardian.config.NettyConfig;
import com.mls.guardian.exception.RemoteSendRequestException;
import com.mls.guardian.exception.RemoteTimeoutException;
import com.mls.guardian.protocol.CodeType;
import com.mls.guardian.protocol.RemoteBody;
import com.mls.guardian.protocol.RemoteCommand;
import com.mls.guardian.protocol.body.SystemErrorBody;
import com.mls.guardian.remoting.executor.ExecutorThreadsMap;
import com.mls.guardian.remoting.executor.RemoteExecutor;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;

/**
 * Created by jl on 15/9/11.
 */
public abstract  class AbstractNetty {
    private static final Logger log = LoggerFactory.getLogger(AbstractNetty.class);

    protected final ConcurrentHashMap<Integer,ResponseFuture>responseTables =
            new ConcurrentHashMap<Integer,ResponseFuture>();
    protected ConcurrentHashMap<Integer,ExecutorThreadsMap>excutorTables
            = new ConcurrentHashMap<Integer,ExecutorThreadsMap>();
    protected NettyConfig nettyConfig = new NettyConfig();

    public AbstractNetty(NettyConfig nettyConfig){
        if(nettyConfig != null)
            this.nettyConfig = nettyConfig;
    }

    protected void handleReciveMessage(ChannelHandlerContext ctx,RemoteCommand remoteCommand) throws Exception {

        final RemoteCommand cmd = remoteCommand;
        if (cmd != null) {
            if(cmd.getRemoteCommandType().equals(RemoteCommandType.REQUEST)){
                log.info("id[{}] code[{}] receive request  body-{} channel-{}",remoteCommand.getRequestId(),remoteCommand.getCode(),remoteCommand.getRemoteBody(),ctx.channel());
                handleRequest(ctx,cmd);
            }else if(cmd.getRemoteCommandType().equals(RemoteCommandType.RESPONSE)){
                log.info("id[{}] code[{}] receive response body-{} channel-{}",remoteCommand.getRequestId(),remoteCommand.getCode(),remoteCommand.getRemoteBody(),ctx.channel());
                handleResponse(ctx, cmd);
            }
        }
    }


    protected void handleRequest(ChannelHandlerContext ctx,RemoteCommand request) throws Exception {
        int code = request.getCode();
        ExecutorThreadsMap map = this.excutorTables.get(code);

        if(map.getExecutorService() != null){
            Runnable run = new Runnable() {
                @Override
                public void run() {
                    try {
                        final RemoteBody responseBody = map.getRemoteExecutor().execute(ctx, request);

                        if (responseBody != null) {
                            RemoteCommand response = RemoteTools.getReponseRemoteCmd(code, responseBody);
                            response.setRequestId(request.getRequestId());

                            try {
                                ctx.writeAndFlush(response);
                            }
                            catch (Throwable e) {
                                log.error("id[{}] code[{}] process request over, but response failed", request.getRequestId(),request.getCode(),e);
                                log.error(String.valueOf(code));
                                log.error(response.toString());
                            }
                        }
                        else {
                            // 收到请求，但是没有返回应答，可能是processRequest中进行了应答，忽略这种情况
                        }
                    }catch (Throwable e) {
                        log.error("id[{}] code[{}] process request exception", request.getRequestId(),request.getCode(),e);
                        log.error(String.valueOf(code));

                        SystemErrorBody systemError = new SystemErrorBody();

                        systemError.setErrorContent(RemoteTools.exceptionSimpleDesc(e));
                        final RemoteCommand response =
                                RemoteTools.getReponseRemoteCmd(CodeType.SYSTEM_ERROR, systemError);

                        response.setRequestId(request.getRequestId());
                        ctx.writeAndFlush(response);
                    }
                }
            };

            map.getExecutorService().execute(run);
        }else{
            try{
                RemoteBody responseBody = null;
                if(map.getRemoteExecutor() != null){
                    log.info("id[{}] code[{}] find executor is[{}]",request.getRequestId(),request.getCode(),map.getRemoteExecutor());
                    responseBody = map.getRemoteExecutor().execute(ctx,request);
                }else{
                    responseBody = new RemoteBody() {
                        @Override
                        public void setErrorContent(String errorContent) {
                            super.setErrorContent(errorContent);
                        }
                    };
                    log.error("id[{}] code[{}] 未找到对应的处理器",request.getRequestId(),request.getCode());
                    responseBody.setErrorContent("未找到对应的处理器 请求码是["+code+"]");
                }

                RemoteCommand response = RemoteTools.getReponseRemoteCmd(code, responseBody);
                response.setRequestId(request.getRequestId());
                ctx.writeAndFlush(response);
            }catch (Exception e){
                log.error("id[{}] code[{}] process request exception", request.getRequestId(),request.getCode(),e);
                log.error(String.valueOf(code));

                SystemErrorBody systemError = new SystemErrorBody();

                systemError.setErrorContent(RemoteTools.exceptionSimpleDesc(e));
                final RemoteCommand response =
                        RemoteTools.getReponseRemoteCmd(CodeType.SYSTEM_ERROR, systemError);

                response.setRequestId(request.getRequestId());
                ctx.writeAndFlush(response);
            }

        }




    }

    protected void handleResponse(ChannelHandlerContext ctx,RemoteCommand remoteCommand){
        final ResponseFuture responseFuture  = responseTables.get(remoteCommand.getRequestId());

        if(responseFuture != null){
            responseFuture.putResponse(remoteCommand);
            responseTables.remove(remoteCommand.getRequestId());
        }

    }


    public void registerExcutor(int code,RemoteExecutor remoteExcutor){
        ExecutorThreadsMap map = new ExecutorThreadsMap(remoteExcutor,null);

        this.excutorTables.put(code,map);
    }

    public void registerExcutorWithExecutorService(int code,RemoteExecutor remoteExcutor,ExecutorService executorService){
        ExecutorThreadsMap map = new ExecutorThreadsMap(remoteExcutor,executorService);

        this.excutorTables.put(code,map);
    }

    /**
     * 定义清理意外错误导致的没有返回的请求 防止线程池被占满
     */
    public void scanResponseTable() {
        Iterator<Map.Entry<Integer, ResponseFuture>> it = this.responseTables.entrySet().iterator();
        while (it.hasNext()) {
            Map.Entry<Integer, ResponseFuture> next = it.next();
            ResponseFuture rep = next.getValue();

            if ((rep.getBeginTimestamp() + rep.getTimeoutMillis() + 1000) <= System.currentTimeMillis()) {
                it.remove();
                try {
                    rep.executeInvokeCallback();
                }
                catch (Throwable e) {
                    log.warn("scanResponseTable, operationComplete Exception", e);
                }
                finally {
//                    rep.release();
                }

                log.warn("remove timeout request, " + rep);
            }
        }
    }


    public RemoteCommand sendMessageSync(Channel channel,final RemoteCommand request,final long timeoutMillis) throws Exception {
        ResponseFuture responseFuture = new ResponseFuture(timeoutMillis);
        responseTables.put(request.getRequestId(),responseFuture);

        try {
            channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture f) throws Exception {
                    log.info("id[{}] code[{}] sendmessagesync'listener get response success is {} ",request.getRequestId(),request.getCode(),f.isSuccess());
                    if (f.isSuccess()) {
                        responseFuture.setSendRequestOK(true);
                        return;
                    } else {
                        responseFuture.setSendRequestOK(false);
                    }

                    log.info("id[{}] code[{}] ChannelFutureListener remove request.getRequestId()", request.getRequestId(),request.getCode());
                    responseTables.remove(request.getRequestId());
                    responseFuture.setCause(f.cause());
                    responseFuture.putResponse(null);
                    log.warn("send a request command to channel <" + channel.remoteAddress() + "> failed.");
                    log.warn(request.toString());
                }
            });

            RemoteCommand reponse = responseFuture.waitResponse();
            if (reponse == null) {
                if (responseFuture.isSendRequestOk()) {
                    throw new RemoteTimeoutException(RemoteTools.getRemoteAddr(channel), timeoutMillis, responseFuture.getCause());
                } else {
                    throw new RemoteSendRequestException(RemoteTools.getRemoteAddr(channel), responseFuture.getCause());
                }
            }
            return reponse;
        }
        finally {
            responseTables.remove(request.getRequestId());
        }
    }


    public abstract void sendMessageOneWay(Channel channel,final RemoteCommand request) throws Exception;
    public abstract void sendMessageOneWay(String addr,final RemoteCommand request) throws Exception;
}
